package com.bw.app.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.bean.TableProcess;
import com.bw.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * @author whd
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private Connection connection;
    private OutputTag<JSONObject> hbaseTag;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public TableProcessFunction(OutputTag<JSONObject> hbaseTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.hbaseTag = hbaseTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override//初始化连接
    public void open(Configuration parameters) throws Exception {
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //value格式: {"db":"","tn":"","before":{},"after":{},"type":""}
    @Override//处理主流数据     可以发现上下文环境是ReadOnlyContext，因为是负责读状态
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        //1.获取并解析数据    将读取到的String字符串转换为TableProcess
        JSONObject jsonObject = JSONObject.parseObject(value);
        String date = jsonObject.getString("after");
        TableProcess tableProcess = JSON.parseObject(date, TableProcess.class);

        //2.检查HBase表是否存在并使用Phoneix建表
        if (TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())){
            //建表语句: create table if not exists db.tn(id varchar primary key,tm_name varchar) xxxx;
            checkTable(tableProcess.getSinkTable(),tableProcess.getSinkColumns(),tableProcess.getSinkPk(),tableProcess.getSinkExtend());
        }

        //3.写入状态广播出去，只有写入到状态中主流才可以使用，如果不写到状态中主流无法从状态获取
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String key = tableProcess.getSourceTable() +"-"+tableProcess.getOperateType();
        //只需要将数据放入到状态中
        broadcastState.put(key,tableProcess);
    }

    //建表语句: create table if not exists db.tn(id varchar primary key,tm_name varchar) xxxx;
    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {

        PreparedStatement preparedStatement = null;
        try {
            if (sinkPk == null){
                sinkPk = "id";
            }

            if (sinkExtend == null){
                sinkExtend = "";
            }

            StringBuffer createTableSQL = new StringBuffer("create table if not exists ")
                    .append(GmallConfig.HBASE_SCHEMA+"."+sinkTable)
                    .append("(");

            String[] split = sinkColumns.split(",");
            for (int i = 0; i < split.length; i++) {

                String field = split[i];
                //判断是否为主键
                if (sinkPk.equals(field)){
                    createTableSQL.append(field +" varchar primary key");
                }else {
                    createTableSQL.append(field +" varchar");
                }

                //判断是否为最后一个字段，如果不是，则添加 ","
                if(i < split.length-1){
                    createTableSQL.append(" ,");
                }
            }

            createTableSQL.append(")").append(sinkExtend);
            //打印建表语句
            System.out.println(createTableSQL);

            //预编译SQL
            preparedStatement = connection.prepareStatement(createTableSQL.toString());
            //执行
            preparedStatement.execute();

        } catch (SQLException e) {
            //e.printStackTrace();
            throw new RuntimeException("Phoenix表"+sinkTable+"建表失败!");
        }finally {
            if (preparedStatement != null){
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Override//处理广播数据    Context负责写状态
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //1.读取状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String key = value.getString("tableName") + "-" + value.getString("type");
        TableProcess tableProcess = broadcastState.get(key);


        if (tableProcess != null){
            JSONObject date = value.getJSONObject("after");
            //2.过滤数据-只留sinkColumns中配置的字段
            filterColumn(date,tableProcess.getSinkColumns());

            //3.分流
            //将输出表/主题信息写入value
            value.put("sinkTable", tableProcess.getSinkTable());
            if (TableProcess.SINK_TYPE_KAFKA.equals(tableProcess.getSinkType())){
                //Kafka数据，写入主流
                out.collect(value);
            }else if(TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())){
                //HBase数据，写入侧输出流
                ctx.output(hbaseTag,value);
            }
        }else{
            System.out.println("该组合key："+key+"不存在！");
        }
    }

    /**
     * @param date  {"id":"11","tm_name":"whd","logo_url":"aaa"}
     * @param sinkColumns   id,tm_name
 *                          {"id":"11","tm_name":"whd"}
     */
    private void filterColumn(JSONObject date, String sinkColumns) {
        String[] fields = sinkColumns.split(",");
        List<String> columns = Arrays.asList(fields);

        /*Iterator<Map.Entry<String, Object>> iterator = date.entrySet().iterator();
        while (iterator.hasNext()){
            if(!columns.contains(iterator.next().getKey())) {
                iterator.remove();
            }
        }*/
        //简写版
        date.entrySet().removeIf(next -> !columns.contains(next.getKey()));
    }
}
